package com.fanshuai.reactive.reactor.core.flux;

import com.google.common.collect.ImmutableList;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.Callable;

@Slf4j
public class FluxTest {
    static void print(Object o) {
        System.out.println(o);
        try {
            Thread.sleep(1000);
        } catch (Exception e) {

        }
    }

    static void flux0() {
        Flux.just("hello", "world")
                .doOnNext(s -> print("doOnNext: " + s)).subscribe(s -> print("subscribe: " + s));
    }

    static void flux1() {
        Flux.just(1, 2, 3).subscribe(FluxTest::print);
        Flux.just("hello", "world").subscribe(System.out::println);
        Flux.fromArray(new Integer[] {1, 2, 3, 5}).subscribe(System.out::println);
        Flux.fromIterable(ImmutableList.of("a", "b", "c")).subscribe(System.out::println);
        Flux.fromStream(Arrays.stream(new Integer[] {2, 4, 6})).subscribe(System.out::println);

        Flux.empty().subscribe(System.out::println);
        Flux.never().subscribe(System.out::println);

        Flux.range(5, 10).subscribe(System.out::println);

        Flux.interval(Duration.ofSeconds(1)).subscribe(System.out::println);
        Flux.interval(Duration.ofSeconds(5), Duration.ofSeconds(1)).subscribe(System.out::println);
    }

    static void fluxGenerate() {
        Flux.generate(synchronousSink -> {
            synchronousSink.next("data=" + 1);
            synchronousSink.complete();
        }).subscribe(System.out::println);

        Random random = new Random();
        Flux.generate((Callable<ArrayList<Integer>>) ArrayList::new, (list, sink) -> {
            int i = random.nextInt();
            list.add(i);

            if (list.size() == 10) {
                sink.complete();
            }
            sink.next(i);
            return list;
        }).subscribe(System.out::println);
    }

    static void fluxCreate() {
        Flux.create(sink -> {
            for (int i = 0; i < 10; i++) {
                sink.next(i);
            }
            sink.complete();
        }).subscribe(System.out::println);
    }

    static void mono1() {
        Mono.just("test").subscribe(System.out::println);
        Mono.fromCallable(() -> {
            return 1;
        }).subscribe(System.out::println);
        Mono.fromSupplier(() -> "hello").subscribe(System.out::println);
        Mono.fromRunnable(() -> {
            System.out.println("haha");
        }).subscribe();

        Mono.create(monoSink -> {
            monoSink.success("success");
        }).subscribe(System.out::println);

        Mono.delay(Duration.ofSeconds(2)).subscribe(System.out::println);
        Mono.justOrEmpty(Optional.of(112)).subscribe(System.out::println);
        Mono.error(new RuntimeException("ex")).doOnError(Throwable::printStackTrace);
    }

    /**
     * 截取若干元素，形成Flux<List<T>>
     */
    static void flowBuffer() {
        //Flux<T> -> Flux<List<T>>
        Flux.range(1, 100).buffer(50).subscribe(System.out::println);
        Flux.range(1, 10).bufferUntil(i -> i % 2 == 0).subscribe(System.out::println);
        Flux.range(1, 10).bufferWhile(i -> i % 2 == 0).subscribe(System.out::println);

        //Flux.interval(Duration.ofMillis(100)).buffer(Duration.ofSeconds(1)).subscribe(System.out::println);

        //Stream<Flux>
        Flux.interval(Duration.ofMillis(100)).buffer(Duration.ofSeconds(1))
                .take(2).toStream().forEach(System.out::println);
    }

    static void fluxFilter() {
        Flux.range(1, 10).filter(i -> i % 2 == 0).subscribe(System.out::println);
    }

    /**
     * 截取若干元素，形成Flux<Flux<T>>
     */
    static void fluxWindow() {
        Flux.range(1, 100).window(20).subscribe(System.out::println);
        Flux.range(1, 100).buffer(20).subscribe(System.out::println);
        Flux.interval(Duration.ofMillis(100)).window(Duration.ofMillis(1001)).take(5).subscribe(flux -> {
            flux.subscribe(System.out::println);
        });
    }

    //Flux<T1>, Flux<T2> -> Flux<Tuple2<T1, T2>>
    static void fluxZipWith() {
        Flux.just("a", "b").zipWith(Flux.just("c", "d")).subscribe(System.out::println);
        Flux.just("a", "b")
                .zipWith(Flux.just("c", "d"), (a, b) -> {
                    return String.format("%s:%s", a, b);
                }).subscribe(System.out::println);
    }

    //取走前面的若干元素
    static void fluxTake() {
        Flux.range(1, 10).take(2).subscribe(System.out::println);
        Flux.range(1, 10).takeLast(2).subscribe(System.out::println);
        Flux.range(1, 10).takeUntil(i -> i > 7).subscribe(System.out::println);
        Flux.range(1, 10).takeUntil(i -> i % 2 == 1).subscribe(System.out::println);
    }

    //计算
    static void fluxReduce() {
        Flux.range(1, 100).reduce(Integer::sum).subscribe(System.out::println);
        Flux.range(1, 100).reduce("0", (s1, s2) -> s1 + s2).subscribe(System.out::println);
        Flux.range(1, 100).reduceWith(() -> "0", (s1, s2) -> s1 + s2).subscribe(System.out::println);
    }

    /**
     * 多个Flux合并为一个
     */
    static void fluxMerge() {
        int n = 32;
        /*Flux.merge(Flux.interval(Duration.ofMillis(0), Duration.ofMillis(100)).take(n),
                Flux.interval(Duration.ofMillis(50), Duration.ofMillis(100)).take(n))
                .subscribe(System.out::println);*/

        Flux.mergeSequential(Flux.interval(Duration.ofMillis(0), Duration.ofMillis(100)).take(n),
                Flux.interval(Duration.ofMillis(50), Duration.ofMillis(100)).take(n))
                .subscribe(System.out::println);
    }

    /**
     * 将Flux中每个元素转换为Flux，并将所有的Flux合成一个Flux
     */
    static void fluxFlatMap() {
        Flux.range(1, 10).flatMap(i -> Flux.range(1, i)).subscribe(System.out::println);
        Flux.range(1, 10).flatMapSequential(i -> Flux.interval(Duration.ofMillis(100)).take(i)).subscribe(System.out::println);
    }

    /**
     * 与mergeSequential类似
     */
    static void fluxConcatMap() {
        //Flux.just(5, 10).concatMap(i -> Flux.range(1, i)).subscribe(System.out::println);
        Flux.just(5, 10).concatMap(i -> Flux.interval(Duration.ofMillis(100)).take(i)).subscribe(System.out::println);
    }

    /**
     * 组合最新的元素
     */
    static void fluxCombineLast() {
        Flux.combineLatest(Arrays::toString,
                Flux.range(1, 10), Flux.range(1, 50)).subscribe(System.out::println);
    }

    static void fluxError() {
        //concatWith  2个Flux合并为1个
        Flux.just(1, 2).concatWith(Flux.just(3)).concatWith(Flux.error(new RuntimeException("aa")))
                .subscribe(System.out::println, System.err::println);

        Flux.just(1, 2).concatWith(Flux.just(3)).concatWith(Flux.error(new RuntimeException("aa")))
                .onErrorReturn(0)
                .subscribe(System.out::println, System.err::println);

        Flux.just(1, 2).concatWith(Flux.just(3)).concatWith(Flux.error(new RuntimeException("aa")))
                .onErrorResume(e -> {
                    if (e instanceof RuntimeException) {
                        return Mono.just(0);
                    }
                    if (e instanceof Exception) {
                        return Mono.just(-1);
                    }
                    return Mono.empty();
                })
                .subscribe(System.out::println, System.err::println);
    }

    static void fluxScheduler() {
        Flux<String> flux = Flux.create(stringFluxSink -> {
            for (int i = 0; i < 4.; i++) {
                stringFluxSink.next(Thread.currentThread().getName());
            }
            stringFluxSink.complete();
        });
        flux.publishOn(Schedulers.single()).map(s -> String.format("[%s] %s", Thread.currentThread().getName(), s))
                .publishOn(Schedulers.elastic()).map(s -> String.format("[%s] %s", Thread.currentThread().getName(), s))
                .subscribeOn(Schedulers.elastic()).subscribe(System.out::println);
    }

    public static void main(String[] args) throws Exception {
        //flux0();
        //flux1();
        //fluxGenerate();
        //fluxCreate();
        //mono1();

        //flowBuffer();
        //fluxFilter();
        //fluxWindow();
        //fluxZipWith();
        //fluxTake();
        //fluxReduce();
        //fluxMerge();
        //fluxFlatMap();
        //fluxConcatMap();
        //fluxCombineLast();
        //fluxError();
        //fluxScheduler();
        Thread.sleep(3600000);
    }
}
